package broadcast

import (
	"encoding/json"
	"fmt"
	"gitee.com/viwii/vconf/db/dao"
	"net"
	"sync"
	"time"
)

type RequestArg struct {
	Method        string //方法，目前仅支持sub和get,和ping两种方法
	AppName       string //产品名
	ClusterName   string //集群名
	NamespaceName string //命名空间名
	MsgID         int64  //消息ID号，多路复用时候返回请求的ID号
}

type ReplyMsg struct {
	Method        string //方法，目前仅支持sub和get,和ping两种方法
	AppName       string //产品名
	ClusterName   string //集群名
	NamespaceName string //命名空间名
	MsgID         int64  //消息ID号，多路复用时候返回请求的ID号
	Status        int    //状态码，一般 0 是正常
	Message       string //消息
	Content       []byte //内容
}

type RegisterInfo struct {
	PkStream PacketStream
	ID int64
}

type NetSvr struct {
	Mutex sync.Mutex
	PubMap    map[string][]RegisterInfo //发布订阅结构
	currentID int                       //最新的id，用于监控数据库是否有最新的发布
}

var defaultNetSvr *NetSvr

func Svr() *NetSvr {
	if defaultNetSvr == nil {
		defaultNetSvr = new(NetSvr)
	}

	return defaultNetSvr
}

func (b *NetSvr) scanAndNotify() {
	for {
		time.Sleep(time.Second)

		newestID, err := dao.GetDao().GetNewestID()
		if err != nil {
			fmt.Errorf("GetNewestMessageID",err)
			continue
		}

		if newestID > b.currentID {
			msgs, err := dao.GetDao().GetReleaseNewestID(b.currentID)

			if err != nil {
				fmt.Errorf("GetReleaseMessage",err)
				continue
			}

			for _, msg := range msgs {
				keyStr := fmt.Sprintf("key_%s_%s_%s", msg.AppName, msg.ClusterName, msg.NamespaceName)

				b.Mutex.Lock()
				if ary,ok:=b.PubMap[keyStr];ok{
					for _, itm := range ary {
						rep := &ReplyMsg{Method: "update",
							MsgID:         itm.ID,
							AppName:       msg.AppName,
							ClusterName:   msg.ClusterName,
							NamespaceName: msg.NamespaceName,
							Content:       []byte(msg.Value),
							Status:        0,
						}
						data, _ := json.Marshal(rep)

						itm.PkStream.Write(&Packet{Data: data})
					}
				}
				b.Mutex.Unlock()

				if int(msg.ID) > b.currentID {
					b.currentID = int(msg.ID)
				}

			}


		}
	}
}


func (n *NetSvr) StartSvr(ipstr string) {
	go n.StartListen(":3456")

	var err error
	n.currentID, err = dao.GetDao().GetNewestID()
	if err != nil {
		fmt.Println("start dao err:",err)
	}

	go n.scanAndNotify()
}

func AcceptRquest(n *NetSvr, conn net.Conn) {
	pk := NewPacketStream(conn, true)
	for {
		// 创建一个新切片， 用作保存数据的缓冲区
		pkg, err := pk.Read()
		if err != nil {
			pk.Close()
			n.Mutex.Lock()

			for keyStr,ary:= range n.PubMap {
				for i := 0; i < len(ary); i++ {
					if ary[i].PkStream.ID() == pk.ID(){
						n.PubMap[keyStr] = append(ary[:i],ary[i+1:]...)
					}
				}
			}

			n.Mutex.Unlock()


			break
		}

		var arg RequestArg

		err = json.Unmarshal(pkg.Data, &arg)
		if err != nil {
			pk.Close()
		}
		fmt.Println(arg.Method, arg.AppName, arg.ClusterName, arg.NamespaceName)

		switch arg.Method {
		case "get":
			dao := dao.GetDao()
			cfg, err := dao.ReadConfig(arg.AppName, arg.ClusterName, arg.NamespaceName)
			if err != nil {
				pk.Close()
			}

			fmt.Println("-----------------------get",arg.AppName,arg.ClusterName,arg.NamespaceName)
			rep := &ReplyMsg{Method: "get",
				MsgID:         arg.MsgID,
				AppName:       arg.AppName,
				ClusterName:   arg.ClusterName,
				NamespaceName: arg.NamespaceName,
				Content:       []byte(cfg.Value),
				Status:        0,
			}
			data, _ := json.Marshal(rep)
			pk.Write(&Packet{Data: data})

			fmt.Println("end")

		case "quicksub":
			keyStr := fmt.Sprintf("key_%s_%s_%s", arg.AppName, arg.ClusterName, arg.NamespaceName)
			n.Mutex.Lock()
			if ary, ok := n.PubMap[keyStr]; ok {
				for _, itm := range ary {
					if itm.PkStream.ID() == pk.ID() {
						goto loopEndNow
					}
				}
				n.PubMap[keyStr] = append(n.PubMap[keyStr], RegisterInfo{
					PkStream: pk,
					ID:arg.MsgID,
				})
			loopEndNow:
			} else {
				var ary []RegisterInfo
				ary = append(ary,RegisterInfo{
					PkStream: pk,
					ID:arg.MsgID,
				})

				n.PubMap[keyStr] = ary
			}
			n.Mutex.Unlock()
		case "sub":
			keyStr := fmt.Sprintf("key_%s_%s_%s", arg.AppName, arg.ClusterName, arg.NamespaceName)
			n.Mutex.Lock()
			if ary, ok := n.PubMap[keyStr]; ok {
				for _, itm := range ary {
					if itm.PkStream.ID() == pk.ID() {
						goto loopEnd
					}
				}
				n.PubMap[keyStr] = append(n.PubMap[keyStr], RegisterInfo{
					PkStream: pk,
					ID:arg.MsgID,
				})
			loopEnd:
			} else {
				var ary []RegisterInfo
				ary = append(ary,RegisterInfo{
					PkStream: pk,
					ID:arg.MsgID,
				})

				n.PubMap[keyStr] = ary
			}

			rep := &ReplyMsg{Method: "sub",
				MsgID:  arg.MsgID,
				Status: 0,
				AppName:       arg.AppName,
				ClusterName:   arg.ClusterName,
				NamespaceName: arg.NamespaceName,
			}

			data, _ := json.Marshal(rep)
			pk.Write(&Packet{Data: data})
			n.Mutex.Unlock()
		case "ping":
			rep := &ReplyMsg{Method: "pong",
				MsgID: arg.MsgID,
				AppName:       arg.AppName,
				ClusterName:   arg.ClusterName,
				NamespaceName: arg.NamespaceName,
			}
			data, _ := json.Marshal(rep)
			pk.Write(&Packet{Data: data})
		case "unsub":
			keyStr := fmt.Sprintf("key_%s_%s_%s", arg.AppName, arg.ClusterName, arg.NamespaceName)
			n.Mutex.Lock()
			if ary, ok := n.PubMap[keyStr]; ok {
				for idx, itm := range ary {
					if itm.PkStream.ID() == pk.ID() {
						n.PubMap[keyStr] = append(n.PubMap[keyStr][:idx], n.PubMap[keyStr][idx+1:]...)
						break
					}
				}

			}

			rep := &ReplyMsg{Method: "unsub",
				MsgID:         arg.MsgID,
				AppName:       arg.AppName,
				ClusterName:   arg.ClusterName,
				NamespaceName: arg.NamespaceName,
				Status:        0,
			}

			n.Mutex.Unlock()
			data, _ := json.Marshal(rep)
			pk.Write(&Packet{Data: data})
		}
	}
}

func (n *NetSvr) StartListen(ipstr string) {
	n.PubMap = make(map[string][]RegisterInfo)
	listen, err := net.Listen("tcp", ipstr) // 创建用于监听的 socket
	if err != nil {
		fmt.Println("listen err=", err)
		return
	}

	defer listen.Close() // 服务器结束前关闭 listener

	// 循环等待客户端来链接
	for {
		conn, err := listen.Accept() // 创建用户数据通信的socket
		if err != nil {
			fmt.Println("Accept() err=", err)
		} else {
			fmt.Println("通信套接字，创建成功。。。")
		}

		//开始监听
		go AcceptRquest(n, conn)
	}
}
